package cn.colins.raft.node;

import cn.colins.raft.LogManage;
import cn.colins.raft.RaftNode;
import cn.colins.raft.enity.*;
import cn.colins.raft.enums.NodeStatusEnums;
import cn.colins.raft.config.ServerConfig;
import cn.colins.raft.exception.RaftRunException;
import cn.colins.raft.rpc.session.RpcSession;
import cn.colins.raft.rpc.session.RpcSessionFactory;
import cn.colins.raft.state.StateMachines;
import cn.colins.raft.utils.ThreadPoolUtils;
import cn.hutool.core.collection.CollectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @Author czl
 * @Description leader 节点操作
 * @Date 2023/7/14 17:53
 * @Param
 * @return
 **/
public class LeaderRaftNode implements RaftNode {

    private static final Logger log = LoggerFactory.getLogger(LeaderRaftNode.class);

    private final static NodeStatusEnums NODE_TYPE = NodeStatusEnums.LEADER;


    @Override
    public ClientResponse clientRequestHandler(ClientRequest command, List<ServerConfig> serverConfigs) throws ExecutionException, InterruptedException {
        LogEntity logEntity = new LogEntity(RaftNodeInfo.getInstance().getCurrentTerm(), 0L, command.getCommand());
        ClientResponse clientResponse = ClientResponse.builder().build();
        clientResponse.setRequestId(command.getRequestId());
        LogManage logManage = RaftNodeInfo.getInstance().getLogManage();
        // 1.收到命令首先自己预提交
        long cacheLogId = logManage.preCommitLog(logEntity);
        // 2.给其他节点下发预处理请求，根据结果判断是否提交
        List<Future<AppendEntriesPreCommitResult>> taskList = new ArrayList<>(serverConfigs.size());
        for (ServerConfig serverConfig : serverConfigs) {
            Future<AppendEntriesPreCommitResult> preCommitResultFuture = ThreadPoolUtils.sendAsyncMsgPool.submit(() -> {
                AppendEntriesPreCommit preCommit = AppendEntriesPreCommit.builder().entries(logEntity).preCommitLogId(cacheLogId).build();

                RpcSession<AppendEntriesPreCommitResult, AppendEntriesPreCommit> preCommitSession =
                        RpcSessionFactory.<AppendEntriesPreCommitResult, AppendEntriesPreCommit>openSession(serverConfig, preCommit);
                return preCommitSession == null ? null : preCommitSession.syncSend(1000L);
            });
            taskList.add(preCommitResultFuture);
        }
        // 3.根据返回的状态判断是否提交持久化  > n/2
        // 4.且日志持久化成功
        long commitLogIndex = 0L;
        if (StateMachines.logPreCommitHandler(taskList, serverConfigs.size()) && (commitLogIndex = logManage.commitLog(cacheLogId)) != 0L) {
            long finalCommitLogIndex = commitLogIndex;
            // 5.给其他节点发送提交请求(单向不需要返回了)
            for (ServerConfig serverConfig : serverConfigs) {
                ThreadPoolUtils.sendAsyncMsgPool.execute(() -> {
                    AppendEntriesCommit commit = AppendEntriesCommit.builder().commitLogId(cacheLogId).commitLogIndex(finalCommitLogIndex).build();
                    RpcSession<Void, AppendEntriesCommit> commitSession = RpcSessionFactory.<Void, AppendEntriesCommit>openSession(serverConfig, commit);
                    if (commitSession != null) {
                        commitSession.asyncSend();
                    }
                });
            }
            // 6.给客户的返回结果
            clientResponse.setCode(200);
            clientResponse.setMsg("SUCCESS");
            return clientResponse;
        }
        clientResponse.setCode(400);
        clientResponse.setMsg("提交被拒绝");
        return clientResponse;
    }

    @Override
    public AppendEntriesPreCommitResult logPreCommitHandler(AppendEntriesPreCommit appendEntriesPreCommit) {
        throw new RaftRunException("leader 不可能收到日志预处理请求");
    }

    @Override
    public void logCommitHandler(AppendEntriesCommit appendEntriesCommit) {
        throw new RaftRunException("leader 不可能收到日志提交请求");
    }

    @Override
    public LogIndexPullResult sendLogPullRequest(List<Long> pullLogIndex) {
        throw new RaftRunException("leader不需要发送Log拉取请求");
    }

    @Override
    public synchronized LogIndexPullResult logPullRequestHandler(LogIndexPull logIndexPull) {
        List<LogEntity> logEntities = new ArrayList<>();
        try {
            LogManage logManage = RaftNodeInfo.getInstance().getLogManage();
            RandomAccessFile file = new RandomAccessFile(RaftNodeInfo.getInstance().getLogPath(), "r");
            for (Long logIndex : logIndexPull.getPullLogIndex()) {
                LogEntity logEntity = logManage.getLogEntityByIndex(logIndex-1, file);
                if (logEntity != null) {
                    logEntities.add(logEntity);
                }
            }
        } catch (FileNotFoundException e) {
            log.error("日志读取失败：{}", e.getMessage(), e);
        }
        LogIndexPullResult logIndexPullResult = new LogIndexPullResult(logEntities);
        logIndexPullResult.setRequestId(logIndexPull.getRequestId());
        return logIndexPullResult;
    }


    @Override
    public boolean callVoteRequest(List<ServerConfig> serverConfigs) {
        // leader 不会发起投票请求
        throw new RaftRunException("leader-发起投票  状态出错啦！！！！！！！！！！");
    }

    @Override
    public RequestVoteResult voteRequestHandler(RequestVoteRPC voteRPC) {
        // leader 有可能收到 候选者的投票申请
        RaftNodeInfo instance = RaftNodeInfo.getInstance();
        RequestVoteResult requestVoteResult = RequestVoteResult.builder().build();
        requestVoteResult.setRequestId(voteRPC.getRequestId());
        // 候选人的任期比我大 而且日志还比我大 说明我已经out了，我需要退位
        if (voteRPC.getTerm() >= instance.getCurrentTerm() && voteRPC.getLastLogIndex() > instance.getLastLogIndex()) {
            // 状态变更
            StateMachines.becomeFollow(voteRPC.getTerm(), voteRPC.getCandidateId(), null);

            requestVoteResult.setTerm(voteRPC.getTerm());
            requestVoteResult.setVoteGranted(true);
            log.info(" {}: 我身为现任leader，我认可你的实力，我下位让贤：{}", instance.getSelf().toString(), voteRPC.getCandidateId());
            return requestVoteResult;
        }
        log.info(" {}: 我身为现任leader，不同你的上任请求：{}", instance.getSelf().toString(), voteRPC.getCandidateId());
        // 否则就不同意，而且你还得给我老实点
        requestVoteResult.setTerm(instance.getCurrentTerm());
        requestVoteResult.setVoteGranted(false);
        requestVoteResult.setLeaderId(instance.getSelf().toString());
        return requestVoteResult;
    }

    @Override
    public boolean callHeartBeatRequest(List<ServerConfig> serverConfigs) throws ExecutionException, InterruptedException {

        if (CollectionUtil.isEmpty(serverConfigs)) {
            StateMachines.becomeFollow(RaftNodeInfo.getInstance().getCurrentTerm(), null, null);
            log.debug(" {}: 只有一个leader，还发什么心跳？", RaftNodeInfo.getInstance().getSelf().toString());
            return false;
        }

        List<Future<HeartBeatResult>> taskList = new ArrayList<>(serverConfigs.size());

        // leader 需要发送心跳 防止网络分区，一旦心跳返回不足 n/2 则自动降级
        for (ServerConfig serverConfig : serverConfigs) {
            Future<HeartBeatResult> heartBeatResultFuture = ThreadPoolUtils.sendAsyncMsgPool.submit(() -> {
                HeartBeatRequest build = HeartBeatRequest.builder()
                        .leaderId(RaftNodeInfo.getInstance().getSelf().toString())
                        .leaderLastCommitIndex(RaftNodeInfo.getInstance().getLastLogIndex())
                        .term(RaftNodeInfo.getInstance().getCurrentTerm()).build();
                RpcSession<HeartBeatResult, HeartBeatRequest> heartBeatRequestRpcSession = RpcSessionFactory.<HeartBeatResult, HeartBeatRequest>openSession(serverConfig, build);
                return heartBeatRequestRpcSession == null ? null : heartBeatRequestRpcSession.syncSend(200L);
            });
            taskList.add(heartBeatResultFuture);
        }
        // 响应结果处理
        return StateMachines.heartBeatResultHandler(taskList, serverConfigs.size());
    }

    @Override
    public HeartBeatResult heartBeatHandler(HeartBeatRequest heartBeatRequest) {
        // leader 有可能接收到其他 leader 的心跳 比如：网络分区，导致这个leader被隔离，其他的节点又选出来了一个leader
        // 上述心跳做了处理，一旦分区leader就自动退位了，所以防止了同时多个leader产生的情况
        throw new RaftRunException("leader 收到了心跳！！！！！！！！！！！");
    }
}
